package com.ptkc.core.canal.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.ptkc.core.canal.error.NoHandlerException;
import com.ptkc.utils.SpringUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * @author: Mr.Jiao
 * @date: 2018年09月29日
 */
@Component
public class EventHandlerSuppor {


    final Collection<EventHandler> handlers;


    public Collection<EventHandler> getHandlers(){
        return handlers;
    }


    public EventHandlerSuppor(){
        Map<String,EventHandler> handlerMap = SpringUtils.getApplicationContext().getBeansOfType(EventHandler.class);
        handlers = handlerMap.values();
    }


    @Async
    public void handle(CanalEntry.Entry entry, CanalEntry.RowChange change) {
        if(handlers.size() > 0) {
            String database = entry.getHeader().getSchemaName();
            String table = entry.getHeader().getTableName();
            handlers.forEach((handler) -> {
                if(handler.getSubscribe(database, table)) {
                    handler.handle(entry, change);
                }
            });
        }else{
            throw new NoHandlerException("There are no empty processors handler.");
        }
    }
}
